package com.chiaching.framwork.config;

import com.chiaching.framwork.core.exception.BusException;
import com.chiaching.framwork.core.handler.mq.DefaultMessageListener;
import com.chiaching.framwork.core.handler.mq.MessageListenerImpl;
import com.chiaching.framwork.domian.bean.RocketMqProperties;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ConcurrentHashMap;

/**
 *
 * 延迟消费 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 对应rocket每个延迟等级的时间
 * @author jiajing.he
 * @date 2021/4/10 23:54
 */
@Configuration
@EnableConfigurationProperties({RocketMqProperties.class})
@ConditionalOnProperty(prefix = "super.rocketmq", value = "isEnable", havingValue = "true",matchIfMissing = false)
@Slf4j
@Setter
public class RocketMqConfig  {

    RocketMqProperties rocketMqProperties;

    ApplicationContext applicationContext;
    /**
     * k -> topic  v->MessageListener
     */
    public static final ConcurrentHashMap<String, DefaultMessageListener> MESSAGE_LISTENERS= new ConcurrentHashMap<>(1 << 3);

    public RocketMqConfig(RocketMqProperties rocketMqProperties, ApplicationContext applicationContext) {
        this.rocketMqProperties = rocketMqProperties;
        this.applicationContext = applicationContext;
    }

    @Bean
    public DefaultMQProducer defaultMQProducer(){
        if (StringUtils.isEmpty(rocketMqProperties.getGroupName()) || StringUtils.isEmpty(rocketMqProperties.getNamesrvAddr())) {
            throw new BusException("rocket 一些重要属性为空!");
        }
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(rocketMqProperties.getGroupName());

        producer.setNamesrvAddr(rocketMqProperties.getNamesrvAddr());
        //自动创建topic
        producer.setCreateTopicKey(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC);

        // 如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
        // producer.setInstanceName(instanceName);
        producer.setMaxMessageSize(rocketMqProperties.getProducerMaxMessageSize());
        producer.setSendMsgTimeout(rocketMqProperties.getProducerSendMsgTimeout());
        // 如果发送消息失败，设置重试次数，默认为2次
        producer.setRetryTimesWhenSendFailed(rocketMqProperties.getProducerRetryTimesWhenSendFailed());

        try {
            producer.start();
            log.info("producer is start ! groupName:{},namesrvAddr:{}", rocketMqProperties.getGroupName(),
                    rocketMqProperties.getNamesrvAddr());
        } catch (MQClientException e) {
            log.error(String.format("producer is error {}", e.getMessage(), e));
            throw new RuntimeException(e);
        }
        return producer;
    }


    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(rocketMqProperties.getGroupName());
        defaultMQPushConsumer.setNamesrvAddr(rocketMqProperties.getNamesrvAddr());
        defaultMQPushConsumer.setConsumeThreadMin(this.rocketMqProperties.getConsumerConsumeThreadMin());
        defaultMQPushConsumer.setConsumeThreadMax(this.rocketMqProperties.getConsumerConsumeThreadMax());
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        // consumer.setMessageModel(MessageModel.CLUSTERING);

        /**
         * 设置一次消费消息的条数，默认为1条
         */
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(this.rocketMqProperties.getConsumerConsumeMessageBatchMaxSize());
        defaultMQPushConsumer.registerMessageListener(new MessageListenerImpl(MESSAGE_LISTENERS));
        defaultMQPushConsumer.start();
        return defaultMQPushConsumer;
    }

}
